package com.fengzai.queue;

import com.fengzai.utils.ZkUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;

import java.util.List;
import java.util.Scanner;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;

/**
 * Created by  峰仔  on 2017-08-10
 * 分布式队列  出队
 */
public class Dequeue {
    public static void main(String[] args) throws Exception {
        Scanner scanner = new Scanner(System.in);
        System.out.println("输入要执行的任务");
        String task = scanner.next();
        //由键盘输入要执行的节点
        execute(task);
    }

    public static void execute(String task) throws Exception {
        if (task == null) {
            return;
        }
        CountDownLatch countDownLatch = new CountDownLatch(1);
        CuratorFramework curatorFramework = ZkUtil.getInstace("");
        List<String> queue = curatorFramework.getChildren().forPath("/queue");
        TreeSet<String> set = new TreeSet<>();
        TreeSet<String> taskSet = new TreeSet<>();
        //排序
        String node = null;
        for (String s : queue) {
            set.add(s);
            //取出当前节点下的任务
            byte[] bytes = curatorFramework.getData().forPath("/queue/" + s);
            String t = new String(bytes);
            taskSet.add(t);
            if (task.equals(t)) {
                //获得当前任务所在的节点
                node = s;
            }
        }
        //是当前要执行的任务就执行 完成后删除该任务所在的节点
        if (task.equals(taskSet.first())) {
            Handle(curatorFramework, countDownLatch, node);
        }
        //否则就获取上一个节点
        SortedSet<String> headSet = set.headSet(node);
        if (headSet.isEmpty()) {
            System.out.println("此任务已经处理过了");
            return;
        } else if (headSet.size() == set.size()) {
            System.out.println("此任务不在此队列");
            //结束程序
            countDownLatch.countDown();
            return;
        }
        // 监听上个节点的变化
        NodeCache nodeCache = new NodeCache(curatorFramework, "/queue/" + headSet.last(), false);
        nodeCache.start(true);
        String finalNode = node;
        nodeCache.getListenable().addListener(() -> {
            ChildData currentData = nodeCache.getCurrentData();
            //删除事件
            if (currentData == null) {
                //上个节点被删除后执行次节点的任务
                Handle(curatorFramework, countDownLatch, finalNode);
                //完成后关闭监听
                nodeCache.close();
            }
        });
        System.out.println("在这里等着处理任务");
        countDownLatch.await();
        nodeCache.close();
    }

    //处理任务
    public static void Handle(CuratorFramework curatorFramework, CountDownLatch countDownLatch, String node) throws Exception {
        //用完删除
        curatorFramework.delete().forPath("/queue/" + node);
        System.out.println("处理完成，删除节点  -->  " + node);
        //-1
        countDownLatch.countDown();
    }
}
